package completablefuture;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;

public class CompletableFutureEngine {

    private final static ExecutorService executorService = Executors.newFixedThreadPool(4);

    /**
     * 创建并行任务并执行
     *
     * @param list            数据源
     * @param function        API调用逻辑
     * @param exceptionHandle 异常处理逻辑
     * @return 处理结果列表
     */
    public static <S, T> List<T> parallelFutureJoin(Collection<S> list, Function<S, T> function,
            Consumer<Throwable> exceptionHandle) {

        List<CompletableFuture<T>> completableFutures = list.stream()
                .map(s -> CompletableFuture.supplyAsync(() -> function.apply(s)))
                .collect(Collectors.toList());

        List<T> results = new ArrayList<>();
        try {
            CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[0])).join();
            for (CompletableFuture<T> completableFuture : completableFutures) {
                results.add(completableFuture.get());
            }
        } catch (Exception e) {
            // exceptionHandle.accept(e);
            exceptionHandle.accept(e.getCause());
        }
        return results;
    }

}
